/**
 * Licensed to the Apache Software Foundation (ASF) under one
 * or more contributor license agreements.  See the NOTICE file
 * distributed with this work for additional information
 * regarding copyright ownership.  The ASF licenses this file
 * to you under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance
 * with the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package cn.edu.thu.laud.objectstore.index;

import java.io.IOException;
import java.math.BigInteger;
import java.nio.ByteBuffer;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;

import org.apache.cassandra.db.marshal.IntegerType;
import org.apache.cassandra.db.marshal.MarshalException;
import org.apache.cassandra.utils.ByteBufferUtil;
import org.apache.cassandra.utils.UUIDGen;


/**
 * 
 * @author zhuoan
 *
 */
public class CompositeUtil
{
	
  public final static String DEFAULT_DYNAMIC_COMPOSITE_ALIASES = "(b=>BytesType,i=>IntegerType,t=>TimeUUIDType,s=>UTF8Type)";
  public final static String[] INDEX_TYPES= {"BytesType", "IntegerType", "TimeUUIDType", "UTF8Type"};

	public enum CompareType {
	    LESS_THAN_EQUAL((byte) -1), EQUAL((byte) 0), GREATER_THAN_EQUAL((byte) 1);

	    private final byte equality;

	    CompareType(byte equality) {
	      this.equality = equality;
	    }

	    public byte toByte() {
	      return equality;
	    }

	    public static CompareType fromByte(byte equality) {
	      if (equality > 0) {
	        return GREATER_THAN_EQUAL;
	      }
	      if (equality < 0) {
	        return LESS_THAN_EQUAL;
	      }
	      return EQUAL;
	    }
	}
	
	public static ByteBuffer createDynamicCompositeKey(Object[] o)
		    throws IOException
    {
		CompareType[] equality = new CompareType[o.length];
		for(int i=0; i<equality.length; i++)
			equality[i] = CompareType.EQUAL;
		 return createDynamicCompositeKey(o, equality);
	}
	
	
  public static ByteBuffer createDynamicCompositeKey(Object[] o, CompareType[] equality)
    throws IOException
  {
    ByteBufferOutputStream out = new ByteBufferOutputStream();
    for (int j = 0; j < o.length; j++) {
      if ((o[j] instanceof String))
      {
        ByteBuffer bytes = ByteBufferUtil.bytes((String)o[j]);
        out.writeShort((short)(0x8000 | 's'));
        out.writeShort((short)bytes.remaining());
        out.write(bytes);
        out.write(equality[j].toByte());
      }
      else if ((o[j] instanceof UUID))
      {
        out.writeShort((short)(0x8000 | 't'));
        out.writeShort((short)16);
        out.write(UUIDGen.decompose((UUID)o[j]));
        out.write(equality[j].toByte());
      }
      else if ((o[j] instanceof ByteBuffer))
      {
        ByteBuffer bytes = ((ByteBuffer)o[j]).duplicate();
        out.writeShort((short)(0x8000 | 'b'));
        out.writeShort((short)bytes.remaining());
        out.write(bytes);
        out.write(equality[j].toByte());
      }
      else if ((o[j] instanceof Number))
      {
    	ByteBuffer bytes = IntegerType.instance.decompose(BigInteger.valueOf(((Number)o[j]).longValue()));
        out.writeShort((short)(0x8000 | 'i'));
        out.writeShort((short)bytes.remaining());
        out.write(bytes);
        out.write(equality[j].toByte());
      }
      else
      {
        throw new MarshalException(o[j].getClass().getName() + " is not recognized as a valid type for this composite");
      }
    }

    return out.getByteBuffer();
  }

  public static List<Object> deserialize(ByteBuffer b) throws IOException
  {
    b = b.duplicate();
    List<Object> components = new ArrayList<Object>();
    while (b.remaining() > 0)
    {
      short header = b.getShort();
      if (header == (short)(0x8000 | 's'))
      {
        ByteBuffer value = getWithShortLength(b);
        components.add(ByteBufferUtil.string(value));
      }
      else if (header == (short)(0x8000 | 't'))
      {
        ByteBuffer value = getWithShortLength(b);
        components.add(UUIDGen.getUUID(value));
      }
      else if (header == (short)(0x8000 | 'b'))
      {
        components.add(getWithShortLength(b));
      }
      else if (header == (short)(0x8000 | 'i'))
      {
        ByteBuffer value = getWithShortLength(b);
    	components.add(IntegerType.instance.compose(value));
      }
      else
      {
        throw new MarshalException(String.valueOf(header & 0xFF) + " is not recognized as a valid type");
      }

      b.get();
    }
    return components;
  }

  protected static ByteBuffer getWithShortLength(ByteBuffer bb)
  {
    int length = getShortLength(bb);
    return getBytes(bb, length);
  }

  protected static int getShortLength(ByteBuffer bb)
  {
    int length = (bb.get() & 0xFF) << 8;
    return length | bb.get() & 0xFF;
  }

  protected static ByteBuffer getBytes(ByteBuffer bb, int length)
  {
    ByteBuffer copy = bb.duplicate();
    copy.limit(copy.position() + length);
    bb.position(bb.position() + length);
    return copy;
  }
}


